package chapter

// Use H2Driver to connect to an H2 database
import scala.slick.driver.MySQLDriver.simple._
import scala.slick.jdbc.meta.MTable
import scala.slick.model.ForeignKeyAction
import scala.slick.jdbc.{ GetResult, StaticQuery => Q }
//import scala.slick.jdbc.JdbcBackend.Database.dynamicSession

object FirstExample extends App {

  import scala.slick.driver.ExtendedProfile
  import Q.interpolation
  // Definition of the SUPPLIERS table
  case class Supplier(id: Int, name: String, street: String, city: String, state: String, zip: String)
  class Suppliers(tag: Tag) extends Table[Supplier](tag, "SUPPLIERS") {
    def id = column[Int]("SUP_ID", O.PrimaryKey) // This is the primary key column
    def name = column[String]("SUP_NAME")
    def street = column[String]("STREET")
    def city = column[String]("CITY")
    def state = column[String]("STATE")
    def zip = column[String]("ZIP")
    // Every table needs a * projection with the same type as the table's type parameter
    def * = (id, name, street, city, state, zip) <> (Supplier.tupled, Supplier.unapply _)
  }
  val suppliers = TableQuery[Suppliers]

  // Definition of the COFFEES table
  case class Coffee(name: String, supID: Int, price: Double, sales: Int, total: Int)
  class Coffees(tag: Tag) extends Table[Coffee](tag, "COFFEES") {
    def name = column[String]("COF_NAME", O.PrimaryKey)
    def supID = column[Int]("SUP_ID")
    def price = column[Double]("PRICE")
    def sales = column[Int]("SALES")
    def total = column[Int]("TOTAL")
    def * = (name, supID, price, sales, total) <> (Coffee.tupled, Coffee.unapply _)
    // A reified foreign key relation that can be navigated to create a join
    def supplier = foreignKey("SUP_FK", supID, suppliers)(_.id, onUpdate = ForeignKeyAction.Restrict, onDelete = ForeignKeyAction.Cascade)
  }
  val coffees = TableQuery[Coffees]
  // Connect to the database and execute the following block within a session
  Database.forURL("jdbc:mysql://127.0.0.1:3306/test?autoReconnect=true&useUnicode=true&characterEncoding=utf-8", "root", "admin", null, driver = "com.mysql.jdbc.Driver") withSession {
    implicit session =>
      // The session is never named explicitly. It is bound to the current
      // thread as the threadLocalSession that we imported
      val tablesInTest = MTable.getTables.list.map(_.name.name)

      //MTable |-> name:MQName |-> name:String  .map{mName => (mName.catalog_?,mName.schema,mName.name) }
      println("MTable |-> name:" + MTable.getTables.list.map(_.name).map { mName => (mName.catalog_?, mName.schema, mName.name) })

      // Create the tables, including primary and foreign keys
      if (tablesInTest.contains("SUPPLIERS".toLowerCase())) {
        (suppliers.ddl ++ coffees.ddl)drop
      }

      (suppliers.ddl ++ coffees.ddl).create

      // Insert some suppliers
      suppliers += Supplier(49, "Superior Coffee", "1 Party Place", "Mendocino", "CA", "95460")
      suppliers.insertAll(
        Supplier(101, "Acme, Inc.", "99 Market Street", "Groundsville", "CA", "95199"),
        Supplier(150, "The High Ground", "100 Coffee Lane", "Meadows", "CA", "93966"))

      // Insert some coffees (using JDBC's batch insert feature, if supported by the DB)
      coffees.insertAll(
        Coffee("Colombian", 101, 7.99, 0, 0),
        Coffee("French_Roast", 49, 8.99, 0, 0),
        Coffee("Espresso", 150, 9.99, 0, 0),
        Coffee("Colombian_Decaf", 101, 8.99, 0, 0),
        Coffee("French_Roast_Decaf", 49, 9.99, 0, 0))

      // Iterate through all coffees and output them
      println("Coffees:")
      coffees foreach {
        case Coffee(name, supID, price, sales, total) =>
          println("  " + name + "\t" + supID + "\t" + price + "\t" + sales + "\t" + total)
      }

      // Why not let the database do the string conversion and concatenation?
      println("Coffees (concatenated by DB):")
      val q1 = for (c <- coffees) // Coffees lifted automatically to a Query
        yield ConstColumn("  ") ++ c.name ++ "\t" ++ c.supID.asColumnOf[String] ++
        "\t" ++ c.price.asColumnOf[String] ++ "\t" ++ c.sales.asColumnOf[String] ++
        "\t" ++ c.total.asColumnOf[String]
      // The first string constant needs to be lifted manually to a ConstColumn
      // so that the proper ++ operator is found
      q1 foreach println

      // Perform a join to retrieve coffee names and supplier names for
      // all coffees costing less than $9.00
      println("Manual join:")
      val q2 = for {
        c <- coffees if c.price < 9.0
        s <- suppliers if s.id === c.supID
      } yield (c.name, s.name)
      for (t <- q2) println("  " + t._1 + " supplied by " + t._2)

      // Do the same thing using the navigable foreign key
      println("Join by foreign key:")
      val q3 = for {
        c <- coffees if c.price < 9.0
        s <- c.supplier
      } yield (c.name, s.name)
      // This time we read the result set into a List
      val l3: List[(String, String)] = q3.list
      for ((s1, s2) <- l3) println("  " + s1 + " supplied by " + s2)

      // Check the SELECT statement for that query
      println(q3.selectStatement)

      // Compute the number of coffees by each supplier
      println("Coffees per supplier:")
      val q4 = (for {
        c <- coffees
        s <- c.supplier
      } yield (c, s)).groupBy(_._2.id).map {
        case (_, q) => (q.map(_._2.name).min.get, q.length)
      }
      // .get is needed because SLICK cannot enforce statically that
      // the supplier is always available (being a non-nullable foreign key),
      // thus wrapping it in an Option
      q4 foreach {
        case (name, count) =>
          println("  " + name + ": " + count)
      }
      
      //Sorting and Filtering
      // limit offset
      val q5 = for (c <- coffees) yield c
      println(q5.drop(3).take(5).selectStatement) //List(Coffee(French_Roast,49,8.99,0,0), Coffee(French_Roast_Decaf,49,9.99,0,0))

      // Inserting

      def insert(c: Coffee) =
        sqlu"INSERT INTO coffees VALUES(${c.name}, ${c.supID}, ${c.price}, ${c.sales}, ${c.total}) ON DUPLICATE KEY UPDATE  price=${c.price}, sales=${c.sales}, total=${c.total}".first

      val coffees_data = Seq(
        Coffee("Colombian", 101, 7.99, 0, 0),
        Coffee("French_Roast", 49, 8.99, 0, 0),
        Coffee("Espresso", 150, 9.99, 0, 0))

      val count = (for (c <- coffees_data) yield insert(c)).sum
      println(s"End insert $count cffees:" + coffees.list)

  }

}